package org.gitos.cloud.comet.tcp;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

import java.net.InetSocketAddress;

import javax.annotation.Resource;

import org.gitos.cloud.domain.comm.Constant;
import org.gitos.cloud.util.ZkUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

/**
 * TcpBootstrap
 * <pre>
 * 置为多例，存储了运行状态
 * 启动TCP服务
 * </pre>
 *
 * @author purely
 * @date 2015/12/4
 */
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
@Component
public class TcpBootstrap implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(TcpBootstrap.class);

    private volatile Boolean state = Constant.STOP_STATE;

    @Autowired
    private ServerBootstrap serverBootstrap;

    @Resource(name = "boss")
    private EventLoopGroup boss;

    @Resource(name = "tcpWorker")
    private EventLoopGroup worker;

    @Value("${" + Constant.NODE_ID + "}")
    private String nodeId;

    @Value("${" + Constant.TCP_PORT + "}")
    private String tcpPort;

    @Value("${" + Constant.TCP_IP + "}")
    private String tcpIp;

    @Autowired
    private ChildChannelInit childChannelInit;


    private void initServer() {
        serverBootstrap.group(boss, worker);
        serverBootstrap.channel(NioServerSocketChannel.class);
        serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
        serverBootstrap.childHandler(childChannelInit);
        serverBootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024);
        serverBootstrap.childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024);
        serverBootstrap.childOption(ChannelOption.SO_RCVBUF, 32 * 1024);
        serverBootstrap.childOption(ChannelOption.SO_SNDBUF, 32 * 1024);
        serverBootstrap.childOption(ChannelOption.SO_BACKLOG, 8 * 128);
        serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
        serverBootstrap.childOption(ChannelOption.SO_REUSEADDR, true);
        serverBootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);
    }

    public void run() {
        log.info("TcpBootstrap running...");
        do {
            synchronized (this) {
                if (state == Constant.RUN_STATE) {
                    break;
                }
            }
            initServer();
            if (StringUtils.isEmpty(tcpIp)){
                tcpIp = "0.0.0.0";
            }
            ChannelFuture bindFuture = serverBootstrap.bind(tcpIp, Integer.valueOf(tcpPort));
            try {
                log.info("开始bind");
                ChannelFuture bindSync = bindFuture.sync();
                log.info("获取bind的channel");
                Channel ch = bindSync.channel();
                InetSocketAddress localAddr = (InetSocketAddress) ch.localAddress();
                log.info("host:{}, port:{}", localAddr.getAddress().getHostAddress(), localAddr.getPort());
                startup();
                log.info("等待关闭boss");
                ch.closeFuture().sync();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            } finally {
                boss.shutdownGracefully();
                worker.shutdownGracefully();
            }
        } while (false);
    }

    public void startup() {
        synchronized (this) {
            if (state == Constant.STOP_STATE) {
                state = Constant.RUN_STATE;
            }
            log.info("通知boss启动完毕");
            this.notifyAll();
            log.info("zookeeper注册");
            if(ZkUtil.registerNodeId(nodeId, tcpIp, tcpPort) == false){
               log.error("zookeeper 注册失败,退出.");
                System.exit(0);
            }
        }
    }

    public void waitForStartup(long timeout) {
        synchronized (this) {
            if (this.state == Constant.STOP_STATE) {
                try {
                    if (timeout == 0) {
                        this.wait();
                    } else {
                        this.wait(timeout);
                    }
                } catch (InterruptedException e) {
                    log.info(e.getMessage(), e);
                }
            }
        }
    }
}
